package com.chencong.source;

import com.chencong.env.FlinkTableEnv;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * @Author chencong
 * @Description 从kafka读取
 * @Date 8:09 下午 2021/8/16
 * @Param
 **/
public class SourceKafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamTableEnvironment = FlinkTableEnv.getStreamTableEnvironment();
        //kafka配置
        Properties properties = new Properties();

        properties.getProperty("bootstrap.servers", "aq2:9092,aq3:9092,aq4:9092");
        properties.getProperty("group.id", "chencong");
        properties.getProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.getProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.getProperty("auto.offset.reset", "latest");

        //读取数据源
        // kafka 的offset 保存在 Source算子的状态
        // kafka不知道offset更新到哪，用一些监控kafka的工具，看不到offset的进展
        // setCommitOffsetsOnCheckpoints设为 true， 会同步一份给kafka，就可以看到了
        DataStreamSource<String> sourceData = streamTableEnvironment.addSource(new FlinkKafkaConsumer<String>("chencong", new SimpleStringSchema(), properties));

        sourceData.print();

        streamTableEnvironment.execute();


    }
}
